/******************************************************************************
* Copyright (C) 2020 ShenZhen Powerdata Information Technology Co.,Ltd
* All Rights Reserved.
* 本软件为***研制。未经本公司正式书面同意，其他任何个人、团体不得使用、
* 复制、修改或发布本软件.
*****************************************************************************/

package com.xxl.job.admin.config.aop;

import java.lang.reflect.Method;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;

import javax.annotation.PostConstruct;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.AfterReturning;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.xxl.job.admin.config.XxlJobClientConfig;
import com.xxl.job.admin.config.XxlJobProducer;
import com.xxl.job.admin.config.XxlJobTestConnectConfig;
import com.xxl.job.listener.common.log.entity.JobListenerRequest;

import lombok.extern.slf4j.Slf4j;

/**
* @Title:aop监听stop start
* @ClassName:  JobListenerClientProducer   
* @Description:TODO  
* @author: xgf
* @date:   2020年2月23日 上午11:35:18   
* @since   JDK1.8
* @history 2020年2月23日 TODO 新建
*/
@Aspect
@Component
@Slf4j
public class JobListenerClientProducer {
	
	/**
	 * true 是开启 false 是关闭监听
	 */
	@Value("${xxl.job.listener.openJobListener:false}")
	private Boolean openJobListener;
	
	@Autowired
	XxlJobClientConfig xxlJobClientConfig;
	
	@Autowired
	XxlJobTestConnectConfig xxlJobTestConnectConfig;
	
	DefaultMQProducer producer = null;
	private ArrayBlockingQueue<Message> queues =  new ArrayBlockingQueue<Message>(30);
	
    @PostConstruct
    public void init() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
    	log.info("openJobListener : [{}], namesrv : [{}] ", openJobListener, xxlJobClientConfig.getNamesrv());
    	
    	//如果关闭了 监听，直接不用初始化了
    	if(!openJobListener) {
    		return;
    	}
    	producer = new XxlJobProducer(xxlJobClientConfig).getProducer();
    	//初始化验证 new Simpl
    	SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    	Message message = new Message(xxlJobTestConnectConfig.getTopic() , xxlJobTestConnectConfig.getTag(), xxlJobClientConfig.getSystemName() + ":" + xxlJobClientConfig.getIp() + ":" +  simpleDateFormat.format(new Date()), "success".getBytes());
    	SendResult send = producer.send(message, xxlJobClientConfig.getTimeout());
    	log.info("mq 初始化结果: [{}]", send);
    }
	
	 /**
     * 定义切入点，切入点为开始任务
     */
    @Pointcut("execution(public * com.xxl.job.admin.controller.JobInfoController.start(..))")
    public void jobStart(){}
    
	 /**
     * 定义切入点，切入点为 停止任务
     */
    @Pointcut("execution(public * com.xxl.job.admin.controller.JobInfoController.pause(..))")
    public void jobStop(){}
    
    
    /**
     * 开启成功 
    * @Title: doAfterReturning
    * @Description: TODO
    * @param @param ret
    * @param @throws Throwable 
    * @return void
    * @throws
     */
    @Around(value = "jobStop()")
    public Object  around(ProceedingJoinPoint pjp) throws Throwable {
    	
    	/**
    	 * 如果是false 就代表不开启监控 跳过
    	 */
    	if(!openJobListener) {
    		Object proceed = pjp.proceed();
    		return proceed;
    	}
    	Object[] args = pjp.getArgs();
    	MethodSignature signature = (MethodSignature) pjp.getSignature();
        Method method = signature.getMethod();     
    	Object proceed = pjp.proceed();
    	Message  message = null;
    	try {
    		
    		//如果queues 为不为空 就代表 有没处理的数据 处理数据的时候 发现异常
    		if(!queues.isEmpty()) {
    			Iterator<Message> iterator = queues.iterator();
    			while(iterator.hasNext()) {
    				message = iterator.next();
    				SendResult send = producer.send(message, xxlJobClientConfig.getTimeout());
    				log.info("消息发出 ： " + send);
    				if(SendStatus.SEND_OK.equals(send.getSendStatus())) {
    					queues.remove(message);
    				}
    				
    			}
    			
    		}
        	
        	
            // 处理完请求，返回内容
            log.info("jobStop Request args : " + args);
            log.info("jobStop Response result : " + proceed);
            JobListenerRequest jobListenerRequest = new JobListenerRequest();
            Date date = new Date();
            SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
            //先赋值方法名
            jobListenerRequest.setBehavior(String.valueOf(method));
            jobListenerRequest.setBehaviorDate(sdf.format(date));
            jobListenerRequest.setId(String.valueOf(args[0]));
            jobListenerRequest.setIp(xxlJobClientConfig.getIp());
            jobListenerRequest.setMsg("停止任务");
            jobListenerRequest.setParams(JSON.toJSONString(proceed));
            jobListenerRequest.setSystemName(xxlJobClientConfig.getSystemName());
            
            
            String key = xxlJobClientConfig .getKeyPrefix() +  xxlJobClientConfig.getSystemName() + ":" + sdf.format(date);
            
            
            message = new Message(xxlJobClientConfig.getTopic() , xxlJobClientConfig.getTag(), key , JSON.toJSONString(jobListenerRequest).getBytes());
            
            if(producer == null) {
        		queues.add(message);
        		return proceed;
        	}
			SendResult send = producer.send(message, xxlJobClientConfig.getTimeout());
			log.info("消息发出 ： " + send);
			if(!SendStatus.SEND_OK.equals(send.getSendStatus())) {
				queues.add(message);
			}
			
			
		}  catch (IllegalStateException e) {
			//代表着mq 连接namesrv异常 放入queue
			log.error(e.toString());
			queues.add(message);
		}catch (MQClientException e) {
			//MQ 客户端连接失败 broker连接失败
			log.error(e.toString());
			queues.add(message);
		}catch (Exception e) {
			log.error(e.toString());
		}finally {
			return proceed;
		}
    }
	
}
